package com.xmim.fulcrum.discover;

import com.xmim.core.constants.MessageCode;
import com.xmim.core.container.AbstractServerContainer;
import com.xmim.core.container.Container;
import com.xmim.core.container.ServerContainer;
import com.xmim.core.entity.Message;
import com.xmim.core.session.EventType;
import com.xmim.core.session.Session;
import com.xmim.core.session.SessionListener;
import com.xmim.core.session.user.UserSession;
import com.xmim.fulcrum.*;
import com.xmim.fulcrum.handler.IMessageHandler;
import com.xmim.fulcrum.node.FulcrumServiceNode;
import com.xmim.fulcrum.node.FulcrumUserNode;
import com.xmim.fulcrum.session.FulcrumNodeBind;
import com.xmim.fulcrum.session.FulcrumSessionManager;
import com.xmim.fulcrum.session.UserSessionsDao;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * IM 节点的ZK协调客户端
 * create by CAOYOU
 **/

@Slf4j
public abstract class AbstractFulcrumDiscover extends AbstractServerContainer<Integer, IMessageHandler<Message<?>>> implements FulcrumDiscover {
    private static final String COUNTER_PATH = "/im/OnlineCounter";
    private static final String MANAGE_PATH = "/im/nodes";
    private static final String PATH_PREFIX = MANAGE_PATH + "/seq-";

    /*本地用户集合*/
    protected final Map<String, FulcrumNodeBind> sessionsCaches = new ConcurrentHashMap<>();

    private FulcrumServiceNode localNode;
    private FulcrumPathCache<FulcrumUserNode> userNodeCache;
    private FulcrumPathCache<FulcrumServiceNode> serviceNodeCache;
    private FulcrumCounter onlineServixeCounter;
    private FulcrumCounter onlineUserCounter;
    private FulcrumSessionManager sessionManager;

    public AbstractFulcrumDiscover(FulcrumFactory factory, ServerContainer parent) {
        super(factory, parent);
    }


    @Override
    public AbstractFulcrumDiscover addHandler(Object handler) {
        this.checkHandler(handler.getClass());
        IMessageHandler<Message<?>> messageHandler = (IMessageHandler<Message<?>>) handler;
        this.addHandler(messageHandler.getType(), messageHandler);
        return this;
    }

    @Override
    public AbstractFulcrumDiscover setSessionManager(FulcrumSessionManager sessionManager) {
        this.sessionManager = sessionManager;
        return this;
    }

    @Override
    public boolean checkHandler(Class handler) {
        return IMessageHandler.class.isAssignableFrom(handler);
    }


    @Override
    public void discover(UserSession session, Message msg) {
        Collection<IMessageHandler<Message<?>>> messageHandlers = this.handlers.values();
        for (IMessageHandler<Message<?>> messageHandler : messageHandlers) {
            messageHandler.handler(session, msg);
        }
    }

    /**
     * 查询会话处理
     *
     * @param session
     * @return
     */
    private UserSession onFindSession(UserSession session) {
        return session;
    }

    /**
     * 添加会话处理
     *
     * @param session
     */
    protected void onAddSession(UserSession session) {
        this.cacheUserSession(session);
        this.notifyOtherNode(session, MessageCode.NTF_USER_ONLINE);
    }

    /**
     * 销毁会话处理
     *
     * @param session
     */
    private void onDestroySession(UserSession session) {
        notifyOtherNode(session, MessageCode.NTF_USER_OFFLINE);
    }

    @Override
    protected void onInit() {
        this.getFactory().getSessionManager().addSessionListener(new SessionListener() {
            @Override
            public Session event(EventType eventType, Session session) {
                UserSession userSession = (UserSession) session;
                switch (eventType) {
                    case FIND:
                        return onFindSession(userSession);
                    case ADDED:
                        onAddSession(userSession);
                        break;
                    case DESTROY:
                        onDestroySession(userSession);
                        break;
                }
                return session;
            }
        });
    }


    @Override
    protected void onBind() {
        try {
            log.info("开始启动服务发现");
            this.localNode = this.createLocalNode();
            this.sessionManager.setLocalNode(localNode);
            this.userNodeCache = this.createUserCache();
            this.serviceNodeCache = this.createServixeCache();
            this.serviceNodeCache.setFulcrumNodeListener(this.sessionManager);
            this.onlineServixeCounter = this.createOnlineServixeCounter();
            this.onlineUserCounter = this.createOnlineUserCounter();
            log.info("启动服务发现完成");
        } catch (Exception e) {
            log.error("启动服务发现异常", e);
        }
    }

    /**
     * 通知其他节点
     *
     * @param session session
     * @param type    类型
     */
    private void notifyOtherNode(UserSession session, int type) {
        Message<FulcrumNodeBind> message = MessageBuilder.buildNodeBindMsg(type, session, this.localNode);
        this.sessionManager.writeAllClient(message);
    }

    /**
     * 创建在线服务统计器
     *
     * @return
     */
    protected abstract FulcrumCounter createOnlineServixeCounter();

    /**
     * 创建在线用户统计器
     *
     * @return
     */
    protected abstract FulcrumCounter createOnlineUserCounter();

    /**
     * 创建服务节点缓存器
     *
     * @return
     */
    protected abstract FulcrumPathCache<FulcrumServiceNode> createServixeCache();

    /**
     * 创建用户节点缓存器
     *
     * @return
     */
    protected abstract FulcrumPathCache<FulcrumUserNode> createUserCache();

    /**
     * 创建本地节点
     *
     * @return
     */
    protected abstract FulcrumServiceNode createLocalNode() throws Exception;


    @Override
    public Session getSession(String sessionId) {
        FulcrumNodeBind nodeBind = null;
        UserSessionsDao sessionsDao = this.getFactory().getUserSessionsDao();
        if (sessionsDao != null) {
            nodeBind = sessionsDao.get(sessionId);
        }
        if (nodeBind == null || nodeBind.getServiceNode() == null) {
            nodeBind = sessionsCaches.get(sessionId);
        }
        if (nodeBind == null || nodeBind.getServiceNode() == null) {
            return this.getFactory().getSessionManager().getSession(sessionId);
        } else {
            return this.sessionManager.getSession(nodeBind.getServiceNode().getId().toString());
        }
    }

    @Override
    public void cacheUserSession(UserSession userSession) {
        String sessionId = userSession.getSessionId();
        FulcrumNodeBind nodeBind;
        if (!sessionsCaches.containsKey(sessionId)) {
            nodeBind = MessageBuilder.buildFulcrumNodeBind(userSession, this.localNode);
            sessionsCaches.put(sessionId, nodeBind);
        } else {
            nodeBind = sessionsCaches.get(sessionId);
            nodeBind.setServiceNode(this.localNode);
        }
    }

    @Override
    public FulcrumSessionManager getSessionManager() {
        return sessionManager;
    }

    @Override
    public FulcrumPathCache<FulcrumUserNode> getUserCache() {
        return userNodeCache;
    }

    @Override
    public FulcrumPathCache<FulcrumServiceNode> getServiceCache() {
        return serviceNodeCache;
    }

    @Override
    public FulcrumCounter getOnlineServixeCounter() {
        return onlineServixeCounter;
    }

    @Override
    public FulcrumCounter getOnlineUserCounter() {
        return onlineUserCounter;
    }

    @Override
    public FulcrumServiceNode getLocalNode() {
        return localNode;
    }


    public Map<String, FulcrumNodeBind> getSessionsCaches() {
        return sessionsCaches;
    }

    @Override
    public FulcrumFactory getFactory() {
        return (FulcrumFactory) super.getFactory();
    }

}
